/*
 * (c) Copyright 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 * [See end of file]
 */

package com.hp.hpl.jena.clusteredtdb.comms;

import java.io.PrintWriter;
import java.util.Iterator;
import java.util.List;

import lib.Log;

import com.hp.hpl.jena.clusteredtdb.DataNodeServer;
import com.hp.hpl.jena.clusteredtdb.base.UtilityFunctions;
import com.hp.hpl.jena.clusteredtdb.pgraph.NodeMask;
import com.hp.hpl.jena.tdb.base.file.Location;
import com.hp.hpl.jena.tdb.base.objectfile.ObjectFile;
import com.hp.hpl.jena.tdb.base.record.Record;
import com.hp.hpl.jena.tdb.index.RangeIndex;
import com.hp.hpl.jena.tdb.index.TupleIndex;
import com.hp.hpl.jena.tdb.index.TupleIndexRecord;
import com.hp.hpl.jena.tdb.nodetable.NodeTableBase;
import com.hp.hpl.jena.tdb.store.FactoryGraphTDB;
import com.hp.hpl.jena.tdb.store.GraphTDB;
import com.hp.hpl.jena.tdb.store.NodeId;

public class LocalVNode {
	
	public static final int IDSize = NodeId.SIZE ;
	public static final int tupleIds = 3;

	TupleIndex[] tupleIndexes ;
	RangeIndex[] indexes ;
	ObjectFile objects;
	
	NodeMask mask;
	
	int vnodeId;
	DataNodeServer dns;
	
	public LocalVNode(int vnodeid, DataNodeServer dns, String dataloc, NodeMask mask) {
		this.vnodeId = vnodeid;
		this.dns = dns;
		this.mask = mask;
		
//		TODO should probably replace this by creating all the indexes manually myself.
		GraphTDB gtdb = FactoryGraphTDB.createGraph(new Location(dataloc+this.vnodeId));
		tupleIndexes = gtdb.getNodeTupleTable().getTupleTable().getIndexes() ;		
		indexes = new RangeIndex[tupleIndexes.length] ;
		for ( int i = 0 ; i < indexes.length ; i++ )
		{
		    if ( ! (tupleIndexes[i] instanceof TupleIndexRecord ) )
		    {
		        Log.warn(getClass(), "Not a record-based index: "+tupleIndexes[i]) ;
		        continue ;
		    }
		    indexes[i] = ((TupleIndexRecord)tupleIndexes[i]).getRangeIndex() ;
		}
		
//		dependent on being able to get into the guts of the node table here.
		objects = ((NodeTableBase)gtdb.getNodeTupleTable().getNodeTable()).getObjects() ;
	}
	
	/*
	 * All these functions are passed a writer rather than returning an array or list
	 * This is because they may produce a lot of data, so it is 
	 * much more efficient to write it out immediately.  Should convert this
	 * to passing a class to which strings are added - more general.
	 */
	
	public void processFind(Iterator<String> reqIterator, PrintWriter out) {
		String indexName = reqIterator.next();
		String str = reqIterator.next();
		byte[] triple = UtilityFunctions.tupleStringToByteArray(str);
		
		Record r = this.getIndex(indexName).find(new Record(triple,null));
		out.println(UtilityFunctions.createSpacedTupleString(
				UtilityFunctions.tupleBytesToStrings(r.getKey())));
	}
	
	public void processFindRange(Iterator<String> reqIterator, PrintWriter out) {
		String indexName = reqIterator.next();
		String str1 = reqIterator.next();
		String str2 = reqIterator.next();
		
		byte[] tuple1 = UtilityFunctions.tupleStringToByteArray(str1);
		byte[] tuple2 = UtilityFunctions.tupleStringToByteArray(str2);

		Iterator<Record> ir = this.getIndex(indexName).iterator(
				new Record(tuple1,null),new Record(tuple2,null));
		while(ir.hasNext()) {
			out.println(UtilityFunctions.createSpacedTupleString(
					UtilityFunctions.tupleBytesToStrings(ir.next().getKey())));
		}
	}
	
	public void processIdToNode(Iterator<String> reqIterator, PrintWriter out) {
		while(reqIterator.hasNext()) {
			long idnum = Long.parseLong(reqIterator.next());
			NodeId id = new NodeId(this.mask.getDiskAddressFromId(idnum));
		//	dependent on being able to get into the guts of the node table here.
			String node = this.objects.read(id);
			out.println(String.valueOf(idnum) + " " + node);
		}
	}
	
	/*
	 * TODO Should encode strings rather than allowing them to span multiple lines
	 * This might already be being done - check this out.  Currently this
	 * implementation is not completely safe.  Since we'll probably change this to a
	 * non-strings-based implementation soon anyway, this is not a big deal.  Probably.
	 */
	public void processWriteNode(Iterator<String> reqIterator, PrintWriter out) {
		while(reqIterator.hasNext()) {
			String nodeString = reqIterator.next();
			long id = objects.write(nodeString).getId();
			id = this.mask.createMaskedId(this.vnodeId, id);
			out.println(id + " " + nodeString);
		}
	}
	
	//TODO 3-tuple specific at the moment.
	public void processWriteRecord(Iterator<String> reqIterator, PrintWriter out) {
		String[] addkey;
		String indexName = reqIterator.next();
		while(reqIterator.hasNext()) {
			addkey = reqIterator.next().split(" ");
			byte[] tuple = UtilityFunctions.tupleStringToByteArray(UtilityFunctions.createSpacedTupleString(addkey));
			if(!this.getIndex(indexName).add(new Record(tuple, null))) {
				out.println(addkey[0] + " " + addkey[1] + " " + addkey[2]);
			}
		}
	}
	
	public void processGetAllNodes(PrintWriter out) {
		List<String> al = objects.all();
		Iterator<String> it = al.iterator();
		while(it.hasNext()) {
			out.println(it.next());
		}
	}
	
	protected RangeIndex getIndex(String indexName) {
	    
	    indexName = ">"+indexName ;
	    for ( TupleIndex idx : tupleIndexes )
	    {
	        // This comes down to ColMap label format.
	        if ( idx.getLabel().endsWith(indexName) )
	            return ((TupleIndexRecord)idx).getRangeIndex() ;
	    }
	    Log.warn(this, "Not found: Index: "+indexName) ;
		return null;
	}

	public void processCloseObject(PrintWriter out) {
		this.objects.close();		
	}

	public void processSyncObject(Iterator<String> reqIterator, PrintWriter out) {
		boolean force = Boolean.parseBoolean(reqIterator.next());
		this.objects.sync(force);		
	}

	public void processCloseRIndex(Iterator<String> reqIterator, PrintWriter out) {
		RangeIndex index = this.getIndex(reqIterator.next());
		index.close();
	}
	
	public void processSyncRIndex(Iterator<String> reqIterator, PrintWriter out) {
		RangeIndex index = this.getIndex(reqIterator.next());
		boolean force = Boolean.parseBoolean(reqIterator.next());
		index.sync(force);
	}
	
	

}

/*
 * (c) Copyright 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. The name of the author may not be used to endorse or promote products
 *    derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
